package com.uuuo.rxandroid;

import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.observables.GroupedObservable;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;

import static rx.Observable.just;

public class MainActivity extends AppCompatActivity {

    private String tag = "****rxandroid****";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
    }


    /**
     * 事件的消费者(观察者)
     * <p>
     * Observer 事件的接收者
     * Subscriber 事件的订阅者
     * 他们的基本使用方式是完全一样的
     *
     * @param view
     */
    public void createObserver(View view) {
        /**
         * Observer 事件的接收者
         * 即观察者，它决定事件触发的时候将有怎样的行为。
         * **/
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onNext(String s) {
                Log.d(tag, "Item: " + s);
            }

            @Override
            public void onCompleted() {
                Log.d(tag, "Completed!");
            }

            @Override
            public void onError(Throwable e) {
                Log.d(tag, "Error!");
            }
        };

        /**
         * Subscriber 事件的订阅者
         * 除了 Observer 接口之外，RxJava 还内置了一个实现了 Observer 的抽象类：Subscriber。
         * Subscriber 对 Observer 接口进行了一些扩展，但他们的基本使用方式是完全一样的：
         */
        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onNext(String s) {
                Log.d(tag, "Item: " + s);
            }

            @Override
            public void onCompleted() {
                Log.d(tag, "Completed!");
            }

            @Override
            public void onError(Throwable e) {
                Log.d(tag, "Error!");
            }
        };
    }


    /**
     * 事件的生产者(被观察者)
     *
     * @param view
     */
    public void createObservable(View view) {

        /**
         * Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。
         *
         *
         * OnSubscribe 会被存储在返回的 Observable 对象中，它的作用相当于一个计划表，
         * 当 Observable 被订阅的时候，OnSubscribe 的 call() 方法会自动被调用，
         * 事件序列就会依照设定依次触发。
         * 这样，由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。
         */
        Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello");
                subscriber.onNext("Hi");
                subscriber.onNext("Aloha");
                subscriber.onCompleted();
            }
        });

        observable.subscribe(new Observer() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Object o) {

            }
        });
    }

    /**
     * 订阅
     */
    public void subscribe() {
//        observable.subscribe(observer);
//// 或者：
//        observable.subscribe(subscriber);
    }


    public void createAction1() {
        Action1<String> onNextAction = new Action1<String>() {
            // onNext()
            @Override
            public void call(String s) {
                Log.d(tag, s);
            }
        };
        Action1<Throwable> onErrorAction = new Action1<Throwable>() {
            // onError()
            @Override
            public void call(Throwable throwable) {
                // Error handling
            }
        };
        Action0 onCompletedAction = new Action0() {
            // onCompleted()
            @Override
            public void call() {
                Log.d(tag, "completed");
            }
        };

//// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
//    observable.subscribe(onNextAction);
//// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
//    observable.subscribe(onNextAction, onErrorAction);
//// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
//    observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
    }

    /**
     * 使用调配器
     *
     * @param view
     */
    public void useScheduler(View view) {

        Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("1");
                subscriber.onNext("2");
                Commutils.Logi("OnSubscribe--------->" + Thread.currentThread().getName());
            }
        }).subscribeOn(Schedulers.io()).
                observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Commutils.Logi(s);
                        Commutils.Logi("Action1--------->" + Thread.currentThread().getName());
                    }
                });

    }


    /**
     * 变换
     * Map int---->String
     */
    public void map(View view) {
        Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("8");
            }
        }).map(new Func1<String, Integer>() {
            @Override
            public Integer call(String s) {
                return Integer.parseInt(s);
            }
        }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                Commutils.Logi(integer + 10 + "");
            }
        });
    }

    /**
     * 变换
     * flatMap int---->Observable<String>
     */
    public void flatMap(View view) {

        ArrayList<ArrayList<String>> hash = new ArrayList<>();
        ArrayList<String> arrayList1 = new ArrayList<>();
        arrayList1.add("11");
        arrayList1.add("12");
        arrayList1.add("13");
        arrayList1.add("14");

        ArrayList<String> arrayList2 = new ArrayList<>();
        arrayList1.add("21");
        arrayList1.add("22");
        arrayList1.add("23");
        arrayList1.add("24");
        hash.add(arrayList1);
        hash.add(arrayList2);

        Observable
                .from(hash).
                flatMap(new Func1<ArrayList<String>, Observable<String>>() {
                    @Override
                    public Observable<String> call(ArrayList<String> strings) {
                        return Observable
                                .from(strings)
                                .map(new Func1<String, String>() {
                                    @Override
                                    public String call(String s) {
                                        return s + "0";
                                    }
                                });
                    }
                }).
                map(new Func1<String, Integer>() {
                    @Override
                    public Integer call(String s) {
                        return Integer.parseInt(s);
                    }
                }).
                subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Commutils.Logi(integer + 10 + "");
                    }
                });
    }

    /**
     * doOnSubscribe
     * 在订阅前执行，可以指定线程
     *
     * @param view
     */
    public void doOnSubscribe(View view) {
        just(1, 2, 3).
                doOnSubscribe(new Action0() {
                    @Override
                    public void call() {
                        Commutils.Logi("doOnSubscribe--->" + Thread.currentThread().getName());
                    }
                }).
                subscribeOn(Schedulers.io()).
                subscribeOn(AndroidSchedulers.mainThread())
                .flatMap(new Func1<Integer, Observable<String>>() {
                    @Override
                    public Observable<String> call(Integer integer) {

                        Commutils.Logi("flatMap--->" + Thread.currentThread().getName());
                        return just(integer + "--");
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Commutils.Logi(s);
                        Commutils.Logi("subscribe--->" + Thread.currentThread().getName());
                    }
                });
    }

    public void subject() {
        Subject subject;
        PublishSubject publishSubject;
    }

    public void publishSubject(View view) {
        final PublishSubject publishSubject = PublishSubject.create();
        publishSubject.onNext("A");
        publishSubject.onNext("B");
        publishSubject.onNext("C");

        Observable observable = Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return publishSubject;
            }
        });

        observable.subscribe(new Action1() {
            @Override
            public void call(Object o) {

            }
        });

        publishSubject.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                Commutils.Logi(o.toString());
            }
        });
    }

    /**
     * interval & time
     * timer()：用于创建Observable，延迟发送一次。（新版）
     * interval()：用于创建Observable，跟TimerTask类似，用于周期性发送。
     * delay()：用于事件流中，可以延迟发送事件流中的某一次发送。
     */
    public void interval(View view) {
        //Subscriber是Long类型，监听到的值是执行的次数
        Observable.interval(3, 1, TimeUnit.SECONDS).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(Long aLong) {
                Commutils.Logi(aLong + "");
            }
        });
    }

    /**
     * 过滤
     *
     * @param view
     */
    public void filter(View view) {
        Integer arr[] = new Integer[]{1, 2, 3, 4, 4, 5, 5, 5, 6};

        Observable.from(arr).
                distinct().
                filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer > 3;
                    }
                }).
                subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Commutils.Logi(integer + "");
                    }
                });
    }

    static class Person {
        public Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        public String name;
        public Integer age;
    }

    /**
     * 分组
     *
     * @param view
     */
    public void groupby(View view) {
        ArrayList arrayList = new ArrayList();
        arrayList.add(new Person("jia", 20));
        arrayList.add(new Person("tao", 24));
        arrayList.add(new Person("ming", 24));
        arrayList.add(new Person("jiatao", 20));
        arrayList.add(new Person("yu", 20));


        Observable.from(arrayList)
                .groupBy(new Func1<Person, Integer>() {

                    @Override
                    public Integer call(Person person) {
                        return person.age;
                    }
                })
                .subscribe(new Observer<GroupedObservable<Integer, Person>>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(final GroupedObservable<Integer, Person> integerPersonGroupedObservable) {
                        integerPersonGroupedObservable.subscribe(new Action1<Person>() {
                            @Override
                            public void call(Person person) {
                                Commutils.Logi("分组：" + integerPersonGroupedObservable.getKey() + "" + person.name);
                            }
                        });

                    }
                });

    }


    public void buffer(View view) {
        Observable.range(1, 6).buffer(5, 1).subscribe(new Observer<List<Integer>>() {
            @Override
            public void onCompleted() {
                Commutils.Logi("-----------------onCompleted:");
            }

            @Override
            public void onError(Throwable e) {
                Commutils.Logi("----------------->onError:");
            }

            @Override
            public void onNext(List<Integer> strings) {
                Commutils.Logi("----------------->onNext:" + strings);
            }
        });
    }

    /**
     * RxJava的	merge()	方法将帮助你把两个甚至更多的
     * Observables合并到他们发射的数据项里。
     *
     * @param view
     */
    public void merge(View view) {
        Observable observable1 = Observable.from(new String[]{"A", "B", "C", "D", "E", "A1", "B1", "C1", "D1", "E1"});
//        Observable observable2 = Observable.from(new Integer[]{1, 2, 3, 4, 5,6,7,8,9});
        Observable observable2 = Observable.interval(1, TimeUnit.MILLISECONDS);


        Observable.merge(observable2, observable1)
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        Commutils.Logi(o.toString());
                    }
                });
    }


    public void zip(View view) {
        Observable observable1 = Observable.from(new String[]{"A", "B", "C", "D", "E", "A1", "B1", "C1", "D1", "E1"});
//        Observable observable2 = Observable.from(new Integer[]{1, 2, 3, 4, 5,6,7,8,9});
        Observable observable2 = Observable.interval(1, TimeUnit.MILLISECONDS);
//
//
//        Observable.zip(observable1, observable2,
//                new Func2<String, Long, String>() {
//
//                    @Override
//                    public String call(String s, Long aLong) {
//                        return s + "-------" + aLong;
//                    }
//                })
//                .subscribe(new Action1<String>() {
//                    @Override
//                    public void call(String o) {
//                        Commutils.Logi(o);
//                    }
//                });

        observable1.join(observable2,
                new Func1() {
                    @Override
                    public Object call(Object o) {
                        return o;
                    }
                }, new Func1() {
                    @Override
                    public Object call(Object o) {
                        return o;
                    }
                }, new Func2() {
                    @Override
                    public Object call(Object o, Object o2) {
                        return o;
                    }
                })
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        Commutils.Logi(o.toString());
                    }
                });
    }

    /**
     * 调度器
     */
    public void schedulers() {
        Schedulers.io().createWorker().schedule(new Action0() {
            @Override
            public void call() {

            }
        });
    }

    public void download(View view) {
        Intent intent = new Intent(MainActivity.this, DownloadActivity.class);
        startActivity(intent);

    }

    boolean isError = true;
    int index = 0;

    public void retryWhen(View view) {
        Observable.just("1", "2", "3")
                .flatMap(new Func1<String, Observable<String>>() {
                    @Override
                    public Observable<String> call(String s) {
                        if (s.equals("3") && !isError) {
                            return Observable.error(new NullPointerException("111"));
                        } else {
                            return Observable.just(s);
                        }
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(Observable<? extends Throwable> observable) {
                        Commutils.Logi("-------------retryWhen");
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable error) {
                                isError = false;
                                // For IOExceptions, we  retry
                                if (error instanceof IllegalArgumentException || error instanceof NullPointerException) {
                                    return Observable.just(null);
                                } else {
                                    return Observable.error(error);
                                }
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        Commutils.Logi("onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Commutils.Logi("onError");
                    }

                    @Override
                    public void onNext(String s) {
                        Commutils.Logi("onNext------>" + s);
                    }
                });
    }
}